package cn.doitedu.kafka.day02

import java.util.{Properties, UUID}

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

/**
 * 演示Producer向Kafka写入数据，并开启事务
 * 事务：可以保证多条数据在一个事务中，写入要成功都成，如果其中一条写入失败，全部回滚或标记脏数据
 *
 * Kafka从0.9版本开始支持事务
 *
 * 可以保证写入到Kafka的数据可以实现精准一致性（Exactly Once）
 *
 * 命令行消费者要加上：--isolation-level read_committed
 * 程序要加上 ： properties.setProperty("isolation.level", "read_committed")
 */
object ProducerTransactionDemo {

  def main(args: Array[String]): Unit = {


    //指定生产者的参数
    val properties = new Properties()
    //broker的地址
    // 连接kafka节点
    properties.setProperty("bootstrap.servers", "node-1.51doit.cn:9092,node-2.51doit.cn:9092,node-3.51doit.cn:9092")
    //指定key序列化方式
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    //指定value序列化方式
    properties.setProperty("value.serializer", classOf[StringSerializer].getName) // 两种写法都行

    //如果开启事务，每次必须设置唯一的事务ID
    //设置事务ID的目录，就是当这次事务中，出现异常，要将此次事务中已经写入到Kafka中数据标记为脏数据
    properties.setProperty("transactional.id", UUID.randomUUID().toString)

    val topic = "test"
    //创建Kafka生产者
    val producer = new KafkaProducer[String, String](properties)

    //KafkaProducer默认没有开启事务
    //初始化事务
    producer.initTransactions()
    //开始事务
    producer.beginTransaction()

    //将数据封装到ProducerRecord中
    val record1 = new ProducerRecord[String, String](topic, "hello spark 88811666")
    val record2 = new ProducerRecord[String, String](topic, "hello spark 88811777")

    //调用send方法，而是将数据缓存到client端
    try {
      producer.send(record1)
      //将客户端的数据发送到Broker中，但是没有提交事务
      //producer.flush() //没有提交事务的数据，属于脏数据
      //认为的制造异常
      //val i = 1000 / 0
      producer.send(record2)
      //调用flush将数据刷过去
      producer.flush()
      //提交事务
      //提交事务时，会将对应事务ID的数据，标记为事务成功
      producer.commitTransaction()
    } catch {
      case e: Exception => {
        //放弃当前的事务，会将客户端缓存的数据，丢弃
        //客户端没法将当前事务，已经flush到Broker中的数据丢弃
        producer.abortTransaction()
      }
    }
    producer.close()











  }

}
